#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <stdbool.h>
#include "MQTTClient.h"
#include "shmem.h"
#include "cJSON.h"
#include "msg_queue_peer.h"

#define ADDRESS "tcp://localhost:1883"
#define CLIENTID "ExampleClientSub"
#define TOPIC_UP "/app/data/up"
#define TOPIC_DOWN "/app/data/down"
#define PAYLOAD "Hello World!"
#define QOS 1
#define TIMEOUT 10000L

#define STM32 3  // STM32的消息队列的消息类型
#define MODBUS 1 // MODBUS的消息队列的消息类型

static int flag_ = 0;

pthread_t tid_datapub;
pthread_t tid_msgctl;
char *submsg[128];
char *pubmsg[128];

// 共享内存相关变量
struct shm_param para;
int *p;
struct node_data *n_data;

cJSON *Pointroot;
cJSON *report;

int flag_type0 = 0; // 刷新上报标志位

int size;
union val_t
{
    bool bval;
    int ival;
    float fval;
};

struct node_data // 共享内存数据存储格式
{
    int key;
    int type;
    int dev_type;
    union val_t old_val;
    union val_t new_val;
    int ret;
};

struct setdata // 消息队列数据格式
{
    long devtype;
    int key;
    int type;
    union val_t val;
};

volatile MQTTClient_deliveryToken deliveredtoken;

void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    char *payloadptr;
    payloadptr = message->payload;
    for (int i = 0; i < message->payloadlen; i++)
    {
        putchar(*payloadptr++);
    }
    putchar('\n');

    // 解析json
    cJSON *root = cJSON_Parse(message->payload);
    cJSON *pubtype = cJSON_GetObjectItem(root, "type");
    int type = pubtype->valueint;

    if (type == 1)
    {
        flag_type0 = 1;
    }
    else if (type == 2)
    {
        cJSON *data = cJSON_GetObjectItem(root, "data");
        cJSON *key = cJSON_GetObjectItem(data, "key");
        printf("%s:%d\n",key->string,key->valueint);
        cJSON *val = cJSON_GetObjectItem(data, "val");
        struct setdata d;
        int flag = 0; 
        for (int i = 0; i < *p; i++)
        {
            if (n_data[i].key == key->valueint)
            {
                flag = 1;
                printf("n_data:%d  key:%d\n",n_data[i].key,key->valueint);
                d.devtype = n_data[i].dev_type;
                
                d.key = key->valueint;
                d.type = n_data[i].type;

                if (n_data[i].type == 1)
                {
                    d.val.bval = atoi(val->valuestring);
                }
                else if (n_data[i].type == 2)
                {
                    d.val.ival = atoi(val->valuestring);
                }
                else
                {
                    d.val.fval = atof(val->valuestring);
                }
                break;
            }
        }
        if (flag == 0)
        {
            printf("No such device!!\n");
            cJSON_Delete(root);
            MQTTClient_freeMessage(&message);
            MQTTClient_free(topicName);
            return 1;
        }
        
        if (d.type == 1)
            printf("msg_queue_send:%ld %d %d %d\n", d.devtype, d.key, d.type, d.val.bval);
        else if (d.type == 2)
            printf("msg_queue_send:%ld %d %d %d\n", d.devtype, d.key, d.type, d.val.ival);
        else
            printf("msg_queue_send:%ld %d %d %.2f\n", d.devtype, d.key, d.type, d.val.fval);

        msg_queue_send("msg", &d, sizeof(struct setdata), 0);
        printf("\n");

        struct setdata b = {};
        if (d.devtype == STM32)
        {
            msg_queue_recv("msg", &b, sizeof(struct setdata), STM32, 0);
            printf("MSGTYPE:STM32\n");
        }
        else if(d.devtype == MODBUS)
        {
            msg_queue_recv("msg", &b, sizeof(struct setdata), MODBUS, 0);
            printf("MSGTYPE:MODBUS\n");
         }

        if (b.type == 1)
            printf("msg_queue_recv:%ld %d %d %d\n", b.devtype,b.key, b.type,b.val.bval);
        else if (b.type == 2)
            printf("msg_queue_recv:%ld %d %d %d\n", b.devtype,b.key,b.type, b.val.ival);
        else
            printf("msg_queue_recv:%ld %d %d %.2f\n", b.devtype,b.key,b.type,b.val.fval);

        for (int i = 0; i < *p; i++)
        {
            if (n_data[i].key == b.key)
            {
                printf("recv:n_data:%d  key:%d\n",n_data[i].key,b.key);
                if (n_data[i].type == 1)
                {
                    n_data[i].old_val.bval = n_data[i].new_val.bval;
                    n_data[i].new_val.bval = b.val.bval;
                    printf("old:%d new:%d\n",n_data[i].old_val.bval , n_data[i].new_val.bval);
                }
                else if (n_data[i].type == 2)
                {
                    n_data[i].old_val.ival = n_data[i].new_val.ival;
                    n_data[i].new_val.ival = b.val.ival;
                    printf("old:%d new:%d\n",n_data[i].old_val.ival,n_data[i].new_val.ival);
                }
                else
                {
                    n_data[i].old_val.fval = n_data[i].new_val.fval;
                    n_data[i].new_val.fval = b.val.fval;
                    printf("old:%.2f new:%.2f\n",n_data[i].old_val.fval,n_data[i].new_val.fval);
                }
                
            }
        }
    }
    else if (type == 3)
    {
        cJSON *data = cJSON_GetObjectItem(root, "data");
        cJSON *change_type = cJSON_GetObjectItem(data, "type");
        printf("模式修改为：");

        cJSON *newreport = cJSON_CreateObject();
        cJSON_AddItemToObject(newreport, "type", cJSON_CreateNumber(change_type->valueint));

        if (change_type->valueint == 2)
        {
            printf("周期上报\n");
            cJSON *change_period = cJSON_GetObjectItem(data, "period");
            cJSON_AddItemToObject(newreport, "period", cJSON_CreateNumber(change_period->valueint));
        }
        else
        {
            if (change_type->valueint == 0)
                printf("刷新上报\n");
            else
                printf("变化上报\n");

            cJSON *tmp = cJSON_GetObjectItem(report, "period");
            cJSON_AddItemToObject(newreport, "period", cJSON_CreateNumber(tmp->valueint));
        }

        cJSON_ReplaceItemInObject(Pointroot, "report", newreport);

        int fd = open("/mnt/config/node.json", O_WRONLY);
        if (fd < 0)
        {
            perror("open fault");
            return -1;
        }
        char tmp[size + 1];
        memset(tmp, 0, sizeof(tmp));

        char *pointlistbuf = cJSON_Print(Pointroot);

        strcpy(tmp, pointlistbuf);
        tmp[size] = '\0';
        // printf("----------------------------------\n");
        // printf("%s\n",tmp);
        write(fd, tmp, size);
        close(fd);
        cJSON_Delete(newreport);
        free(pointlistbuf);
        sleep(0.5);
        printf("修改成功,请重新启动服务器！\n");
    }

    cJSON_Delete(root);
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}
int main(int argc, char *argv[])
{
    MQTTClient client;
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    int rc;
    int ch;
    int fd; // 点表文件描述符
    char buf[2048];
    int num_data = 0; // 数据点个数
    int pubtype;      // 上报方式 0:刷新上报 1：变化上报 2：周期上报
    int period = 0;   // 上报周期
    int time = 0;     // 计时
    char mqtt_addr[14] = {};
    int mqtt_port;

    // 初始化共享内存
    shm_init(&para, "home", 1024);
    // 获取共享内存地址
    p = shm_getaddr(&para);
    n_data = (struct node_data *)(p + 1);

    printf("共享内存创建完成\n");
    sleep(1);

    // 解析点表
    fd = open("/mnt/config/node.json", O_RDWR);
    if (fd == 0)
    {
        perror("open fault");
        return -1;
    }
    size = lseek(fd, 0, SEEK_END);

    lseek(fd, 0, SEEK_SET);
    int ret = read(fd, buf, size);
    // printf("%s\n", buf);

    Pointroot = cJSON_Parse(buf);
    cJSON *version = cJSON_GetObjectItem(Pointroot, "version");
    printf("%s:%s\n", version->string, version->valuestring);

    report = cJSON_GetObjectItem(Pointroot, "report");
    cJSON *Pointlist_type = cJSON_GetObjectItem(report, "type");
    pubtype = Pointlist_type->valueint;
    cJSON *Pointlist_period = cJSON_GetObjectItem(report, "period");
    period = Pointlist_period->valueint;

    cJSON *mqtt_server = cJSON_GetObjectItem(Pointroot, "mqtt_server");
    cJSON *list_mqtt_addr = cJSON_GetObjectItem(mqtt_server, "addr");
    strcpy(mqtt_addr, list_mqtt_addr->valuestring);
    cJSON *list_mqtt_period = cJSON_GetObjectItem(mqtt_server, "port");
    mqtt_port = list_mqtt_period->valueint;

    cJSON *stm32 = cJSON_GetObjectItem(Pointroot, "stm32");
    cJSON *stm32_data = cJSON_GetObjectItem(stm32, "data");
    num_data += cJSON_GetArraySize(stm32_data);

    cJSON *modbus = cJSON_GetObjectItem(Pointroot, "modbus");
    cJSON *modbus_data = cJSON_GetObjectItem(modbus, "data");
    num_data += cJSON_GetArraySize(modbus_data);
    printf("点表解析完成\n");
    sleep(1);
    *p = num_data;

    for (int i = 0; i < cJSON_GetArraySize(stm32_data); i++)
    {
        cJSON *item = cJSON_GetArrayItem(stm32_data, i);
        cJSON *key = cJSON_GetObjectItem(item, "key");
        n_data[i].key = key->valueint;
        cJSON *type = cJSON_GetObjectItem(item, "type");
        n_data[i].type = type->valueint;
        n_data[i].dev_type = 3;
    }

    for (int i = cJSON_GetArraySize(stm32_data); i < *p; i++)
    {
        cJSON *item = cJSON_GetArrayItem(modbus_data, i - cJSON_GetArraySize(stm32_data));
        cJSON *key = cJSON_GetObjectItem(item, "key");
        n_data[i].key = key->valueint;
        cJSON *type = cJSON_GetObjectItem(item, "type");
        n_data[i].type = type->valueint;

        n_data[i].dev_type = 1;
    }

    // for (int i = 0; i < *p; i++)
    // {
    //     printf("%d %d %d %f\n",n_data[i].key,n_data[i].type,n_data[i].dev_type,n_data[i].new_val.fval);
    // }

    printf("共享内存写入完成\n");
    sleep(1);

    MQTTClient_create(&client, ADDRESS, CLIENTID,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);

    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }

    MQTTClient_subscribe(client, TOPIC_DOWN, QOS);
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    MQTTClient_deliveryToken token;

    // 组json
    char *ptr = NULL;
    printf("当前上报模式为：");
    if (pubtype == 0) // 变化上报
        printf("刷新上报\n");
    else if (pubtype == 1) // 变化上报
        printf("变化上报\n");
    else
    {
        printf("周期上报\n");
        printf("周期：%d\n",period);
    }
    int i = 0;
    int a = 0;
    int flag = 0;
    static int needflush=0;
    while (1)
    {
        sleep(0.1);

        cJSON *root = cJSON_CreateObject();
        cJSON_AddNumberToObject(root, "type", 1);
        cJSON_AddNumberToObject(root, "result", 0);
        cJSON *data = cJSON_CreateArray();
        if (pubtype == 0||(pubtype==1&&needflush==0)) // 刷新上报
        {
            if (flag_type0 == 1||pubtype==1&&needflush==0)
            {
                for (int i = 0; i < *p; i++)
                {
                    if (n_data[i].type == 1)
                    {
                        // printf("%d %d %d\n",i,n_data[i].type,n_data[i].key);
                        cJSON *item = cJSON_CreateObject();
                        cJSON_AddNumberToObject(item, "key", n_data[i].key);
                        char buf[32];
                        sprintf(buf, "%d", n_data[i].new_val.bval);
                        cJSON_AddStringToObject(item, "val", buf);
                        cJSON_AddItemToArray(data, item);
                    }
                    else if (n_data[i].type == 2)
                    {
                        // printf("%d %d %d\n",i,n_data[i].type,n_data[i].key);
                        cJSON *item = cJSON_CreateObject();
                        cJSON_AddNumberToObject(item, "key", n_data[i].key);
                        char buf[32];
                        sprintf(buf, "%d", n_data[i].new_val.ival);
                        cJSON_AddStringToObject(item, "val", buf);
                        cJSON_AddItemToArray(data, item);
                    }
                    else
                    {
                        // printf("%d %d %d\n",i,n_data[i].type,n_data[i].key);
                        cJSON *item = cJSON_CreateObject();
                        cJSON_AddNumberToObject(item, "key", n_data[i].key);
                        char buf[32];
                        sprintf(buf, "%.2f", n_data[i].new_val.fval);
                        cJSON_AddStringToObject(item, "val", buf);
                        cJSON_AddItemToArray(data, item);
                    }
                   
                }
                printf("全部数据上报完成\n");
                cJSON_AddItemToObject(root, "data", data);
                ptr = cJSON_Print(root);
                pubmsg.payload = ptr;
                pubmsg.payloadlen = (int)strlen(ptr);
                pubmsg.qos = QOS;
                pubmsg.retained = 0;

                MQTTClient_publishMessage(client, TOPIC_UP, &pubmsg, &token);
                flag_type0 = 0;
                free(ptr);
            }
        }
        else if (pubtype == 1) // 变化上报
        {
            i++;
            if (i == 0)
            {
                for (int i = 0; i < *p; i++)
                {
                    if (n_data[i].type == 1)
                    {
                        // printf("%d %d %d\n",i,n_data[i].type,n_data[i].key);
                        cJSON *item = cJSON_CreateObject();
                        cJSON_AddNumberToObject(item, "key", n_data[i].key);
                        char buf[32];
                        sprintf(buf, "%d", n_data[i].new_val.bval);
                        cJSON_AddStringToObject(item, "val", buf);
                        cJSON_AddItemToArray(data, item);
                    }
                    else if (n_data[i].type == 2)
                    {
                        // printf("%d %d %d\n",i,n_data[i].type,n_data[i].key);
                        cJSON *item = cJSON_CreateObject();
                        cJSON_AddNumberToObject(item, "key", n_data[i].key);
                        char buf[32];
                        sprintf(buf, "%d", n_data[i].new_val.ival);
                        cJSON_AddStringToObject(item, "val", buf);
                        cJSON_AddItemToArray(data, item);
                    }
                    else
                    {
                        // printf("%d %d %d\n",i,n_data[i].type,n_data[i].key);
                        cJSON *item = cJSON_CreateObject();
                        cJSON_AddNumberToObject(item, "key", n_data[i].key);
                        char buf[32];
                        sprintf(buf, "%.2f", n_data[i].new_val.fval);
                        cJSON_AddStringToObject(item, "val", buf);
                        cJSON_AddItemToArray(data, item);
                    }
                }
                cJSON_AddItemToObject(root, "data", data);
                ptr = cJSON_Print(root);
                pubmsg.payload = ptr;
                pubmsg.payloadlen = (int)strlen(ptr);
                pubmsg.qos = QOS;
                pubmsg.retained = 0;

                MQTTClient_publishMessage(client, TOPIC_UP, &pubmsg, &token);
                flag_type0 = 0;
                free(ptr);
            }

            sleep(1);
            for (int i = 0; i < *p; i++)
            {
                cJSON *item = cJSON_CreateObject();
                char buf[32];
                if (n_data[i].type == 1)
                {
                    if (n_data[i].old_val.bval != n_data[i].new_val.bval)
                    {
                        flag = 1;

                        cJSON_AddNumberToObject(item, "key", n_data[i].key);
                        sprintf(buf, "%d", n_data[i].new_val.bval);
                        cJSON_AddStringToObject(item, "val", buf);
                        cJSON_AddItemToArray(data, item);
                        n_data[i].old_val.bval = n_data[i].new_val.bval;
                        printf("%d\n", n_data[i].key);
                        printf("old val:%d\n", n_data[i].old_val.bval);
                        printf("new val:%d\n", n_data[i].new_val.bval);
                    }
                }
                else if (n_data[i].type == 2)
                {
                    if (n_data[i].old_val.ival != n_data[i].new_val.ival)
                    {
                        flag = 1;
                        cJSON *item = cJSON_CreateObject();
                        cJSON_AddNumberToObject(item, "key", n_data[i].key);
                        char *buf;
                        sprintf(buf, "%d", n_data[i].new_val.ival);
                        cJSON_AddStringToObject(item, "val", buf);
                        cJSON_AddItemToArray(data, item);
                        n_data[i].old_val.ival = n_data[i].new_val.ival;
                        printf("%d\n", n_data[i].key);
                        printf("old val:%d\n", n_data[i].old_val.ival);
                        printf("new val:%d\n", n_data[i].new_val.ival);
                    }
                }
                else
                {
                    if (n_data[i].old_val.fval != n_data[i].new_val.fval)
                    {
                        flag = 1;
                        cJSON *item = cJSON_CreateObject();
                        cJSON_AddNumberToObject(item, "key", n_data[i].key);
                        char buf[32];
                        sprintf(buf, "%f", n_data[i].new_val.fval);
                        cJSON_AddStringToObject(item, "val", buf);
                        cJSON_AddItemToArray(data, item);
                        n_data[i].old_val.fval = n_data[i].new_val.fval;
                        printf("%d\n", n_data[i].key);
                        printf("old val:%f\n", n_data[i].old_val.fval);
                        printf("new val:%f\n", n_data[i].new_val.fval);
                    }
                }
            }
            if (flag == 1)
            {
                flag = 0;
                cJSON_AddItemToObject(root, "data", data);
                printf("有数据变化！\n");
                ptr = cJSON_Print(root);
                printf("%s\n", ptr);
                pubmsg.payload = ptr;
                pubmsg.payloadlen = (int)strlen(ptr);
                pubmsg.qos = QOS;
                pubmsg.retained = 0;
                MQTTClient_publishMessage(client, TOPIC_UP, &pubmsg, &token);
                free(ptr);
            }
            else
            {
                if (i % 5 == 0)
                    printf("无数据变化\n");
            }
        }
        else if (pubtype == 2) // 周期上报
        {
            sleep(1);
            printf("等待上报中。。。。\n");
            time++;
            if (time == period)
            {
                time = 0;
                for (int i = 0; i < *p; i++)
                {
                    if (n_data[i].type == 1)
                    {
                        // printf("%d %d %d\n",i,n_data[i].type,n_data[i].key);
                        cJSON *item = cJSON_CreateObject();
                        cJSON_AddNumberToObject(item, "key", n_data[i].key);
                        char buf[32];
                        sprintf(buf, "%d", n_data[i].new_val.bval);
                        cJSON_AddStringToObject(item, "val", buf);
                        cJSON_AddItemToArray(data, item);
                    }
                    else if (n_data[i].type == 2)
                    {
                        // printf("%d %d %d\n",i,n_data[i].type,n_data[i].key);
                        cJSON *item = cJSON_CreateObject();
                        cJSON_AddNumberToObject(item, "key", n_data[i].key);
                        char buf[32];
                        sprintf(buf, "%d", n_data[i].new_val.ival);
                        cJSON_AddStringToObject(item, "val", buf);
                        cJSON_AddItemToArray(data, item);
                    }
                    else
                    {
                        // printf("%d %d %d\n",i,n_data[i].type,n_data[i].key);
                        cJSON *item = cJSON_CreateObject();
                        cJSON_AddNumberToObject(item, "key", n_data[i].key);
                        char buf[32];
                        sprintf(buf, "%.2f", n_data[i].new_val.fval);
                        cJSON_AddStringToObject(item, "val", buf);
                        cJSON_AddItemToArray(data, item);
                    }
                }
                cJSON_AddItemToObject(root, "data", data);
                ptr = cJSON_Print(root);
                printf("%s\n", ptr);
                pubmsg.payload = ptr;
                pubmsg.payloadlen = (int)strlen(ptr);
                pubmsg.qos = QOS;
                pubmsg.retained = 0;
                MQTTClient_publishMessage(client, TOPIC_UP, &pubmsg, &token);
                free(ptr);
            }
        }
        needflush++;
        cJSON_Delete(root);
    }

    close(fd);
    cJSON_Delete(Pointroot);
    MQTTClient_unsubscribe(client, TOPIC_DOWN);
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    shm_del(&para);
    return rc;
}
